package com.luxf.base.domainevent.rocketmq;

import com.luxf.base.domainevent.AbstractDomainEventPublisher;
import com.luxf.base.domainevent.codec.MessageEncoder;
import com.luxf.base.domainevent.model.DomainEventNotice;
import com.luxf.base.domainevent.rocketmq.utils.TopicUtils;
import com.luxf.base.domainevent.utils.EventNoticeUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.support.MessageBuilder;

import javax.annotation.Resource;

/**
 * 基于rocketmq实现领域通知
 */
@Slf4j
public class RocketDomainEventNoticePublisher extends AbstractDomainEventPublisher {

    @Resource
    private RocketMQTemplate rocketMQTemplate;

    @Resource
    private MessageEncoder messageEncoder;

    @Override
    protected void doPublish(DomainEventNotice notice) {
        log.info("经由rocketmq发出领域事件通知");
        Message<?> message = MessageBuilder.withPayload(messageEncoder.encode(notice))
                .copyHeaders(notice.getHeader())
                .build();

        try {
            // 生成topic主题
            String topic = TopicUtils.uniqueTopic(EventNoticeUtils.getClassType(notice));
            rocketMQTemplate.asyncSendOrderly(topic, message,
                    EventNoticeUtils.getAggregateId(notice),
                    new SendCallback() {
                        @Override
                        public void onSuccess(SendResult sendResult) {
                            eventNoticeManager.markSuccess(notice);
                        }

                        @Override
                        public void onException(Throwable throwable) {
                            eventNoticeManager.markFail(notice);
                        }
                    });
        } catch (MessagingException ex) {
            log.error("MessagingException:{}", EventNoticeUtils.getMessageId(notice), ex);
            eventNoticeManager.markFail(notice);
        }
    }

}
